{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "56ea28e3-c67e-4177-a6f3-39395c221e3e",
   "metadata": {},
   "source": [
    "## A03 Move Data to the Exploitation Zone\n",
    "\n",
    "This code will perform data processing and cleaning"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e0228301-8b03-470f-82a1-05a8989a0d03",
   "metadata": {},
   "source": [
    "### 1. init spark session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "c048f9f9-e40c-466a-87db-ceb6c148601a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "c16a71dc-69df-40ba-b06a-f66d4d319fbb",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName(\"unemployment data\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f63fe62e-8424-46ea-9e7b-0a0520157f58",
   "metadata": {},
   "source": [
    "### 2. read the parquet file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "1ef8ec7a-d7b8-43ac-9af5-0b4f5ca0d659",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------+--------------+----------------+----------------+---+--------------------+-------------+------+-----+---+\n",
      "| Any|Codi_Barri|Codi_Districte|Demanda_ocupacio|Demanda_ocupació|Mes|           Nom_Barri|Nom_Districte|Nombre| Sexe|_id|\n",
      "+----+----------+--------------+----------------+----------------+---+--------------------+-------------+------+-----+---+\n",
      "|2016|         1|             1|            NULL|  Atur registrat|  1|            el Raval| Ciutat Vella|  2431|Homes|  1|\n",
      "|2016|         2|             1|            NULL|  Atur registrat|  1|      el Barri Gòtic| Ciutat Vella|   588|Homes|  2|\n",
      "|2016|         3|             1|            NULL|  Atur registrat|  1|      la Barceloneta| Ciutat Vella|   637|Homes|  3|\n",
      "|2016|         4|             1|            NULL|  Atur registrat|  1|Sant Pere, Santa ...| Ciutat Vella|   878|Homes|  4|\n",
      "|2016|         5|             2|            NULL|  Atur registrat|  1|       el Fort Pienc|     Eixample|   693|Homes|  5|\n",
      "|2016|         6|             2|            NULL|  Atur registrat|  1|  la Sagrada Família|     Eixample|  1154|Homes|  6|\n",
      "|2016|         7|             2|            NULL|  Atur registrat|  1|la Dreta de l'Eix...|     Eixample|   772|Homes|  7|\n",
      "|2016|         8|             2|            NULL|  Atur registrat|  1|l'Antiga Esquerra...|     Eixample|   855|Homes|  8|\n",
      "|2016|         9|             2|            NULL|  Atur registrat|  1|la Nova Esquerra ...|     Eixample|  1255|Homes|  9|\n",
      "|2016|        10|             2|            NULL|  Atur registrat|  1|         Sant Antoni|     Eixample|   922|Homes| 10|\n",
      "+----+----------+--------------+----------------+----------------+---+--------------------+-------------+------+-----+---+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "file_path = \"./Formatted Zone/unemployment.parquet\"\n",
    "\n",
    "df = spark.read.parquet(file_path)\n",
    "\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "cd3c7c04-9a13-44d4-a2cd-3cb8d57666e8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Any: string (nullable = true)\n",
      " |-- Codi_Barri: string (nullable = true)\n",
      " |-- Codi_Districte: string (nullable = true)\n",
      " |-- Demanda_ocupacio: string (nullable = true)\n",
      " |-- Demanda_ocupació: string (nullable = true)\n",
      " |-- Mes: string (nullable = true)\n",
      " |-- Nom_Barri: string (nullable = true)\n",
      " |-- Nom_Districte: string (nullable = true)\n",
      " |-- Nombre: string (nullable = true)\n",
      " |-- Sexe: string (nullable = true)\n",
      " |-- _id: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d385039d-1b35-4df7-8842-f4cd5c24a3de",
   "metadata": {},
   "source": [
    "### 3. Explore the data in the 'Any' and Mes column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "83668ce1-f7e5-43f7-9678-81955ee32b32",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---+\n",
      "| Any|Mes|\n",
      "+----+---+\n",
      "|2018|  1|\n",
      "|2017|  1|\n",
      "|2012|  1|\n",
      "|2011|  1|\n",
      "|2013|  1|\n",
      "|2020|  1|\n",
      "|2022|  1|\n",
      "|2023|  1|\n",
      "|2016|  1|\n",
      "|2014|  1|\n",
      "|2021|  5|\n",
      "|2019|  1|\n",
      "|2021|  1|\n",
      "+----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"Any\", \"Mes\").distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "66ef11ff-ea80-43a5-83e2-8e5859e2d21f",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = (\n",
    "    df\n",
    "    .withColumn(\"Any\", df[\"Any\"].cast(\"int\"))\n",
    "    .withColumn(\"Mes\", df[\"Mes\"].cast(\"int\"))\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5266ac2a-5d23-48bc-b715-5d87ba1b549a",
   "metadata": {},
   "source": [
    "### 4. Explore the data in the District field"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "87796c17-6038-44be-8d9d-9d96ece657b4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+----------+-------------------+--------------------+\n",
      "|Codi_Districte|Codi_Barri|      Nom_Districte|           Nom_Barri|\n",
      "+--------------+----------+-------------------+--------------------+\n",
      "|             3|        12|     Sants-Montjuïc|la Marina del Pra...|\n",
      "|             8|        52|         Nou Barris|      la Prosperitat|\n",
      "|             8|        49|         Nou Barris|           Canyelles|\n",
      "|            10|        71|         Sant Martí|Provençals del Po...|\n",
      "|             2|         9|           Eixample|la Nova Esquerra ...|\n",
      "|             7|        35|     Horta-Guinardó|         el Guinardó|\n",
      "|             5|        27|Sarrià-Sant Gervasi|el Putxet i el Farró|\n",
      "|             8|        44|         Nou Barris|Vilapicina i la T...|\n",
      "|            10|        67|         Sant Martí|la Vila Olímpica ...|\n",
      "|             3|        17|     Sants-Montjuïc|       Sants - Badal|\n",
      "|             3|        18|     Sants-Montjuïc|               Sants|\n",
      "|            10|        73|         Sant Martí| la Verneda i la Pau|\n",
      "|             3|        11|     Sants-Montjuïc|        el Poble Sec|\n",
      "|             5|        24|Sarrià-Sant Gervasi|     les Tres Torres|\n",
      "|             9|        59|        Sant Andreu|       el Bon Pastor|\n",
      "|             5|        26|Sarrià-Sant Gervasi|Sant Gervasi - Ga...|\n",
      "|             6|        31|             Gràcia|   la Vila de Gràcia|\n",
      "|             2|         5|           Eixample|       el Fort Pienc|\n",
      "|             8|        46|         Nou Barris| el Turó de la Peira|\n",
      "|             2|        10|           Eixample|         Sant Antoni|\n",
      "+--------------+----------+-------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"Codi_Districte\", \"Codi_Barri\", \"Nom_Districte\", \"Nom_Barri\").distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "d79ca46c-45c5-4e4d-baaf-fc97ef3a7c45",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-------------------+\n",
      "|Codi_Districte|      Nom_Districte|\n",
      "+--------------+-------------------+\n",
      "|             1|       Ciutat Vella|\n",
      "|            10|         Sant Martí|\n",
      "|             2|           Eixample|\n",
      "|             3|     Sants-Montjuïc|\n",
      "|             4|          Les Corts|\n",
      "|             5|Sarrià-Sant Gervasi|\n",
      "|             6|             Gràcia|\n",
      "|             7|     Horta-Guinardó|\n",
      "|             8|         Nou Barris|\n",
      "|             9|        Sant Andreu|\n",
      "|            99|          No consta|\n",
      "+--------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"Codi_Districte\", \"Nom_Districte\").distinct().sort(\"Codi_Districte\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "2ec2d93a-0410-436d-b1c5-a9751a0b4c25",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+\n",
      "|Codi_Barri|           Nom_Barri|\n",
      "+----------+--------------------+\n",
      "|         1|            el Raval|\n",
      "|        10|         Sant Antoni|\n",
      "|        11|        el Poble Sec|\n",
      "|        12|la Marina del Pra...|\n",
      "|        13|   la Marina de Port|\n",
      "|        14|la Font de la Gua...|\n",
      "|        15|         Hostafrancs|\n",
      "|        16|          la Bordeta|\n",
      "|        17|       Sants - Badal|\n",
      "|        18|               Sants|\n",
      "|        19|           les Corts|\n",
      "|         2|      el Barri Gòtic|\n",
      "|        20|la Maternitat i S...|\n",
      "|        21|           Pedralbes|\n",
      "|        22|Vallvidrera, el T...|\n",
      "|        23|              Sarrià|\n",
      "|        24|     les Tres Torres|\n",
      "|        25|Sant Gervasi - la...|\n",
      "|        26|Sant Gervasi - Ga...|\n",
      "|        27|el Putxet i el Farró|\n",
      "+----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"Codi_Barri\", \"Nom_Barri\").distinct().sort(\"Codi_Barri\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d63bf12a-0608-4a0c-bd3e-82ddaa3a8063",
   "metadata": {},
   "source": [
    "#### cast code field to int type"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "765aa0d8-f87f-42ea-8223-226a2b48641e",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = (\n",
    "    df\n",
    "    .withColumn(\"Codi_Districte\", df[\"Codi_Districte\"].cast(\"int\"))\n",
    "    .withColumn(\"Codi_Barri\", df[\"Codi_Barri\"].cast(\"int\"))\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "80c47dec-4fa4-48b8-8c9d-b44d40c54236",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Any: integer (nullable = true)\n",
      " |-- Codi_Barri: integer (nullable = true)\n",
      " |-- Codi_Districte: integer (nullable = true)\n",
      " |-- Demanda_ocupacio: string (nullable = true)\n",
      " |-- Demanda_ocupació: string (nullable = true)\n",
      " |-- Mes: integer (nullable = true)\n",
      " |-- Nom_Barri: string (nullable = true)\n",
      " |-- Nom_Districte: string (nullable = true)\n",
      " |-- Nombre: string (nullable = true)\n",
      " |-- Sexe: string (nullable = true)\n",
      " |-- _id: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "30358afd-92ba-4462-a296-30e5cf5c9327",
   "metadata": {},
   "source": [
    "### 5. the Demanda_ocupacio field"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "d80117bc-58e6-40f7-ba23-d59ef537b4ac",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+------------------+\n",
      "|  Demanda_ocupacio|  Demanda_ocupació|\n",
      "+------------------+------------------+\n",
      "|Demanda No Aturats|              NULL|\n",
      "|    Atur Registrat|              NULL|\n",
      "|              NULL|Demanda no aturats|\n",
      "|Demanda no aturats|              NULL|\n",
      "|              NULL|    Atur registrat|\n",
      "|    Atur registrat|              NULL|\n",
      "+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"Demanda_ocupacio\", \"Demanda_ocupació\").distinct().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cd0733ea-4b27-4140-8807-049533ff6ba2",
   "metadata": {},
   "source": [
    "#### merge two filed, if not null then using the value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "8df2e488-e99d-40b8-97d1-6dc9a3e15e2f",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import coalesce"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "f4b0a9c5-524d-4661-acfc-9d6eaf92a78c",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\n",
    "    \"Demanda_ocupacio\",\n",
    "    coalesce(df[\"Demanda_ocupacio\"], df[\"Demanda_ocupació\"])\n",
    ").drop(\"Demanda_ocupació\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "71588bdd-9e02-4b4a-92cf-835073967729",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|  Demanda_ocupacio|\n",
      "+------------------+\n",
      "|Demanda No Aturats|\n",
      "|    Atur Registrat|\n",
      "|Demanda no aturats|\n",
      "|    Atur registrat|\n",
      "+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"Demanda_ocupacio\").distinct().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1c157ae-cac3-474a-a568-a04589376759",
   "metadata": {},
   "source": [
    "#### map to lower case"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "ef9caa02-7973-43b8-ac90-04d735c3df96",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import lower"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "8b0ae1c5-110c-4a3b-9f1f-fafef8c5f27f",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"Demanda_ocupacio\", lower(df[\"Demanda_ocupacio\"]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "f97c948b-2e2c-4e61-8f7b-82a5d03aefa4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+-----+\n",
      "|  Demanda_ocupacio|count|\n",
      "+------------------+-----+\n",
      "|demanda no aturats|  207|\n",
      "|    atur registrat|  993|\n",
      "+------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupby(\"Demanda_ocupacio\").count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "517aee29-5445-47b8-8b65-f7cbe47941ea",
   "metadata": {},
   "source": [
    "### 6. the mes field"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "edd2d99f-e8ce-4d0f-a92c-89fd215b54cb",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+\n",
      "|Mes|count|\n",
      "+---+-----+\n",
      "|  1| 1199|\n",
      "|  5|    1|\n",
      "+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupby(\"Mes\").count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "859b9680-7fda-45f3-8a8b-ba8a88d28883",
   "metadata": {},
   "source": [
    "### 7. the Sexe field"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "4ad00eee-d82a-4572-9c7a-fd7502ae792a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "| Sexe|count|\n",
      "+-----+-----+\n",
      "|Dones|  403|\n",
      "|Homes|  797|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupby(\"Sexe\").count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3edfec0d-3242-4d74-9852-0b3714344de8",
   "metadata": {},
   "source": [
    "### 8. the Nombre feild"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "1ab2250e-4a41-43ca-a8e3-91a7afff68e5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|            Nombre|\n",
      "+-------+------------------+\n",
      "|  count|              1200|\n",
      "|   mean| 566.2558333333334|\n",
      "| stddev|477.82589580818353|\n",
      "|    min|                 0|\n",
      "|    max|               997|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"Nombre\").describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "33e90b95-3259-47e6-a483-0c21e86bba5c",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"Nombre\", df[\"Nombre\"].cast(\"int\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7afc5c63-6e8c-4a2c-9b6f-dc6e5546269a",
   "metadata": {},
   "source": [
    "### 9. save to parquet file and csv file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "6fac4817-5259-47fd-9cf0-e0fa6a7ab96b",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_path = \"./Exploitation Zone/unemployment.parquet\"\n",
    "\n",
    "df.write.mode(\"overwrite\").parquet(output_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "6293c3fd-68c7-43ad-a1fe-550ecb9c27ac",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_path = \"./Exploitation Zone/unemployment.csv\"\n",
    "\n",
    "df.write.mode(\"overwrite\").csv(output_path, header=True)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
